package com.bff.gaia.mix.api.xjoin.client.hbase.async.rowkeydealer;

import com.bff.gaia.mix.api.functions.async.ResultFuture;
import com.bff.gaia.mix.api.xjoin.client.hbase.core.utils.HbaseUtils;
import com.bff.gaia.mix.api.xjoin.core.enums.ECacheContentType;
import com.bff.gaia.mix.api.xjoin.core.enums.EJoinType;
import com.bff.gaia.mix.api.xjoin.core.side.CacheMissVal;
import com.bff.gaia.mix.api.xjoin.core.side.FieldInfo;
import com.bff.gaia.mix.api.xjoin.core.side.SideInfo;
import com.bff.gaia.mix.api.xjoin.core.side.cache.AbsSideCache;
import com.bff.gaia.mix.api.xjoin.core.side.cache.CacheObj;
import com.bff.gaia.shaded.guava18.com.google.common.collect.Lists;
import com.bff.gaia.shaded.guava18.com.google.common.collect.Maps;
import com.bff.gaia.types.Row;

import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;

import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.filter.SubstringComparator;
import org.apache.hadoop.hbase.filter.ValueFilter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

public class QualifierQueryModeDealer extends AbsRowKeyModeDealer {
	private static final Logger LOG = LoggerFactory.getLogger(QualifierQueryModeDealer.class);

	public QualifierQueryModeDealer(Map<String, String> colRefType, String[] colNames, HTable hBaseClient,
	                                boolean openCache, SideInfo sideInfo) {
		super(colRefType, colNames, hBaseClient, openCache, sideInfo);
	}

	@Override
	public void asyncGetData(String tableName, String rowKeyStr, Row input, ResultFuture<Row> resultFuture,
	                         AbsSideCache sideCache) {
//		System.out.println("(xjoinInfo.getEqualFieldList().get(0))"+(xjoinInfo.getEqualFieldList().get(0)));
//		System.out.println("(String) input.getField(xjoinInfo.getEqualValIndex().get(0))"+input.getField(xjoinInfo.getEqualValIndex().get(0)));
		try {
			Filter valueFilter = new SingleColumnValueFilter(tableName.getBytes(),//family
				(sideInfo.getEqualFieldList().get(0)).getBytes(),//qualifier
				CompareFilter.CompareOp.EQUAL,
				((String) input.getField(sideInfo.getEqualValIndex().get(0))).getBytes()//value
			);
			Scan scan = new Scan();
			scan.setFilter(valueFilter);
			ResultScanner scanner = hBaseClient.getScanner(scan);
			dealOneRow(scanner, rowKeyStr, input, resultFuture, sideCache);

		} catch (IOException e) {
			dealFail("", input, resultFuture);
			e.printStackTrace();
		}
	}


	private String dealOneRow(ResultScanner scanner, String rowKeyStr, Row input, ResultFuture<Row> resultFuture, AbsSideCache sideCache) {
		Iterator<Result> it = scanner.iterator();
		List<Object> cacheContent = Lists.newArrayList();
		int count=0;
		while (it.hasNext()) {
			Result result = it.next();
			count++;
			Map<String, Object> args=getResultMap(result);
			List<Object> sideVal = Lists.newArrayList();
			for (String key : colNames) {
				Object val = args.get(key);
				if (val == null) {
					System.out.println("can't get data with column " + key);
					LOG.error("can't get data with column " + key);
				}
				sideVal.add(val);
			}
			Row row = fillData(input, sideVal);
			if (openCache) {
				cacheContent.add(sideVal);
			}
			resultFuture.complete(Collections.singleton(Row.project(row, outFieldIndexs)));
		}



		if (count == 0) {
			dealMissKey(input, resultFuture);
			if (openCache) {
				sideCache.putCache(rowKeyStr, CacheMissVal.getMissKeyObj());
			}
		}

		if (openCache) {
			sideCache.putCache(rowKeyStr, CacheObj.buildCacheObj(ECacheContentType.MultiLine, cacheContent));
		}

		return "";
	}

	private String dealFail(Object arg2, Row input, ResultFuture<Row> resultFuture) {
		LOG.error("record:" + input);
		LOG.error("get side record exception:" + arg2);
		resultFuture.complete(null);
		return "";
	}


}